package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import static org.apache.flink.api.common.typeinfo.Types.*;

//静态导入：把其他类中的静态属性或方法，通过静态导入的方式导入到当前类中，好比当前类拥有了静态的属性和方法，直接调用


/**
 *
 *
 *  returns(Class c): 只针对定义的POJO类型
 *          如果是Tuple，无效的！可以使用
 *
 *  returns(TypeHint t)
 *      或
 *    returns(TypeInformation<T> typeInfo)
 */
public class Demo7_TupleLamda
{
    public static void main(String[] args) throws Exception {

        System.out.println("hahahaha------------------------");

        //1.获取编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //全局只让一个Task运算
        env.setParallelism(1);

        //2.读数据，获取编程模型
        DataStreamSource<String> dataStreamSource = env.socketTextStream("hadoop103",8888);

        //3.进行处理
         dataStreamSource.flatMap( (String inputLine, Collector<Tuple2<String,Integer>> collector) -> {
            String[] words = inputLine.split(" ");

            for (String word : words) {
                collector.collect(Tuple2.of(word, 1));
            }
        })
         //.returns(new TypeHint<Tuple2<String, Integer>>() {})
           .returns(TUPLE(STRING,INT))
         .keyBy( ele -> ele.f0)
         .sum(1)
         .print();


        //5.触发运行
        env.execute();

    }
}
